#ifndef _M_BROKERSERVER_H_
#define _M_BROKERSERVER_H_

#include "exchange.hpp"
#include "msgqueue.hpp"
#include "binding.hpp"
#include "message.hpp"
#include "virtualhost.hpp"
#include "routing.hpp"
#include "consumer.hpp"
#include "channel.hpp"
#include "connect.hpp"

#include "../mqcommon/threadpool.hpp"
#include "../mqcommon/mqProto.pb.h"
#include "../mqcommon/mqlink.pb.h"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/helper.hpp"

#include "../mqthird/include/muduo/net/TcpServer.h"
#include "../mqthird/include/muduo/net/TcpConnection.h"
#include "../mqthird/include/muduo/net/EventLoop.h"
#include "../mqthird/include/muduo/proto/codec.h"
#include "../mqthird/include/muduo/proto/dispatcher.h"
#include "../mqthird/include/muduo/base/Mutex.h"
#include "../mqthird/include/muduo/base/Logging.h"



namespace MyRabbitMQ
{
    #define DBFILE "meta.db"
    #define HOSTNAME "myVirtualHost"
    #define DEFAULT_THREAD 8
    class BrokerServer
    {
    public:
        BrokerServer(int port,const std::string& vbaseDir)
        :m_loop()
        ,m_server(&m_loop,muduo::net::InetAddress("0.0.0.0",port),"broker_server",muduo::net::TcpServer::kReusePort)
        ,m_dispatcher(std::bind(&BrokerServer::onUnknownMessage,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3))
        ,m_codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage,
        &m_dispatcher,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3)))
        ,m_pool(std::make_shared<threadpool>(DEFAULT_THREAD))
        ,m_connections(std::make_shared<ConnectionManager>())
        ,m_consumers(std::make_shared<ConsumerManager>())
        ,m_host(std::make_shared<VirtualHost>(HOSTNAME,DBFILE,vbaseDir))
        {
            //通过读取历史消息, 添加到消费者管理结构中
            //读取所有历史队列
            MsgQueueMap map = m_host->selectMsgQueue();

            //读取所有历史队列的历史消息, 初始化队列的消费者管理结构
            for(auto& f: map)
            {
                m_consumers->declareQueueConsumer(f.first);
            }

            //注册业务处理函数
            m_dispatcher.registerMessageCallback<protoMQ::openChannelRequest>(std::bind(&BrokerServer::onOpenChannel,this,
            std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            
            m_dispatcher.registerMessageCallback<protoMQ::closeChannelRequest>(std::bind(&BrokerServer::onCloseChannel,this,
            std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
        
            m_dispatcher.registerMessageCallback<protoMQ::declareExchangeRequest>(std::bind(&BrokerServer::onDeclareExchange,this,
            std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            
            m_dispatcher.registerMessageCallback<protoMQ::deleteExchangeRequest>(std::bind(&BrokerServer::onDeleteExchange,this,
            std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            
            m_dispatcher.registerMessageCallback<protoMQ::declareQueueRequest>(std::bind(&BrokerServer::onDeclareQueue,this,
            std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            
            m_dispatcher.registerMessageCallback<protoMQ::deleteQueueRequest>(std::bind(&BrokerServer::onDeleteQueue,this,
            std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            
            m_dispatcher.registerMessageCallback<protoMQ::bindRequest>(std::bind(&BrokerServer::onBind,this,
            std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            
            m_dispatcher.registerMessageCallback<protoMQ::unBindRequest>(std::bind(&BrokerServer::onUnBind,this,
            std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            
            m_dispatcher.registerMessageCallback<protoMQ::queueConsumeRequest>(std::bind(&BrokerServer::onQueueConsume,this,
            std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            
            m_dispatcher.registerMessageCallback<protoMQ::queueCancelConsumeRequest>(std::bind(&BrokerServer::onQueueCancelConsume,this,
            std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            
            m_dispatcher.registerMessageCallback<protoMQ::basicPublishRequest>(std::bind(&BrokerServer::onBasicPublish,this,
            std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));

            m_dispatcher.registerMessageCallback<protoMQ::basicAckRequest>(std::bind(&BrokerServer::onBasicAck,this,
            std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));

            //注册服务器消息到来的处理逻辑, 以及链接到来的处理逻辑
            m_server.setMessageCallback(std::bind(&ProtobufCodec::onMessage,m_codec.get(),std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));

            m_server.setConnectionCallback(std::bind(&BrokerServer::onConnection,this,std::placeholders::_1));
        }

        /////////////////////////////////////////////////////////////////////////////////
        ///各项请求
        /////////////////////////////////////////////////////////////////////////////////

        //打开信道, 创建信道需要在特定链接下创建, 所以需要指定链接
        void onOpenChannel(const muduo::net::TcpConnectionPtr& conn,const openChannelRequestPtr& req,muduo::Timestamp)
        {
            //从链接管理模块中获取特定链接
            Connection::ptr connPtr = m_connections->getConnection(conn);
            
            //判断链接是否为有效链接
            if(connPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:连接为无效连接" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            } 
            connPtr->openChannel(req);
            LOG(logLevel::DEBUG) << "执行完毕" << std::endl;
        }
        
        //关闭信道
        void onCloseChannel(const muduo::net::TcpConnectionPtr& conn,const closeChannelRequestPtr& req,muduo::Timestamp)
       {
            //从链接管理模块中获取特定链接
            Connection::ptr connPtr = m_connections->getConnection(conn);
            
            //判断链接是否为有效链接
            if(connPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:连接为无效连接" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            } 
            connPtr->closeChannel(req);
            LOG(logLevel::DEBUG) << "执行完毕" << std::endl;
       }

       //创建交换机
        void onDeclareExchange(const muduo::net::TcpConnectionPtr& conn,const declareExchangeRequestPtr& req,muduo::Timestamp)
        {
            //从链接管理模块中获取特定链接
            Connection::ptr connPtr = m_connections->getConnection(conn);
            
            //判断链接是否为有效链接
            if(connPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:连接为无效连接" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }

            Channel::ptr channelPtr = connPtr->getChannel(req->channel_id());

            //判断信道是否为有效信道
            if(channelPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:信道为无效信道" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }

            channelPtr->declareExchange(req);
            LOG(logLevel::DEBUG) << "执行完毕" << std::endl;
        }
        
        //删除交换机
        void onDeleteExchange(const muduo::net::TcpConnectionPtr& conn,const deleteExchangeRequestPtr& req,muduo::Timestamp)
        {
            //从链接管理模块中获取特定链接
            Connection::ptr connPtr = m_connections->getConnection(conn);
            
            //判断链接是否为有效链接
            if(connPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:连接为无效连接" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }

            Channel::ptr channelPtr = connPtr->getChannel(req->channel_id());

            //判断信道是否为有效信道
            if(channelPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:信道为无效信道" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }
            
            channelPtr->deleteExchange(req);
            LOG(logLevel::DEBUG) << "执行完毕" << std::endl;
        }

        //创建队列
        void onDeclareQueue(const muduo::net::TcpConnectionPtr& conn,const declareQueueRequestPtr& req,muduo::Timestamp)
        {
            //从链接管理模块中获取特定链接
            Connection::ptr connPtr = m_connections->getConnection(conn);
            
            //判断链接是否为有效链接
            if(connPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:连接为无效连接" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }

            Channel::ptr channelPtr = connPtr->getChannel(req->channel_id());

            //判断信道是否为有效信道
            if(channelPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:信道为无效信道" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }

            channelPtr->declareQueue(req);
            LOG(logLevel::DEBUG) << "执行完毕" << std::endl;
        }
        
        //删除队列
        void onDeleteQueue(const muduo::net::TcpConnectionPtr& conn,const deleteQueueRequestPtr& req,muduo::Timestamp)
        {
            //从链接管理模块中获取特定链接
            Connection::ptr connPtr = m_connections->getConnection(conn);
            
            //判断链接是否为有效链接
            if(connPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:连接为无效连接" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }

            Channel::ptr channelPtr = connPtr->getChannel(req->channel_id());

            //判断信道是否为有效信道
            if(channelPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:信道为无效信道" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }
            
            channelPtr->deleteQueue(req);
            LOG(logLevel::DEBUG) << "执行完毕" << std::endl;
        }

        //创建绑定
        void onBind(const muduo::net::TcpConnectionPtr& conn,const bindRequestPtr& req,muduo::Timestamp)
        {
            //从链接管理模块中获取特定链接
            Connection::ptr connPtr = m_connections->getConnection(conn);
            
            //判断链接是否为有效链接
            if(connPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:连接为无效连接" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }

            Channel::ptr channelPtr = connPtr->getChannel(req->channel_id());

            //判断信道是否为有效信道
            if(channelPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:信道为无效信道" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }
            
            channelPtr->bind(req);
            LOG(logLevel::DEBUG) << "执行完毕" << std::endl;
        }

        //删除绑定
        void onUnBind(const muduo::net::TcpConnectionPtr& conn,const unBindRequestPtr& req,muduo::Timestamp)
        {
            //从链接管理模块中获取特定链接
            Connection::ptr connPtr = m_connections->getConnection(conn);
            
            //判断链接是否为有效链接
            if(connPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:连接为无效连接" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }

            Channel::ptr channelPtr = connPtr->getChannel(req->channel_id());

            //判断信道是否为有效信道
            if(channelPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:信道为无效信道" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }
            
            channelPtr->unBind(req);
            LOG(logLevel::DEBUG) << "执行完毕" << std::endl;
        }
        
        //队列订阅
        void onQueueConsume(const muduo::net::TcpConnectionPtr& conn,const queueConsumeRequestPtr& req,muduo::Timestamp)
        {
            //从链接管理模块中获取特定链接
            Connection::ptr connPtr = m_connections->getConnection(conn);
            
            //判断链接是否为有效链接
            if(connPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:连接为无效连接" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }

            Channel::ptr channelPtr = connPtr->getChannel(req->channel_id());

            //判断信道是否为有效信道
            if(channelPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:信道为无效信道" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }
            
            channelPtr->queueConsume(req);
            LOG(logLevel::DEBUG) << "执行完毕" << std::endl;
        }

        //取消队列订阅
        void onQueueCancelConsume(const muduo::net::TcpConnectionPtr& conn,const queueCancelConsumeRequestPtr& req,muduo::Timestamp)
        {
            //从链接管理模块中获取特定链接
            Connection::ptr connPtr = m_connections->getConnection(conn);
            
            //判断链接是否为有效链接
            if(connPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:连接为无效连接" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }

            Channel::ptr channelPtr = connPtr->getChannel(req->channel_id());

            //判断信道是否为有效信道
            if(channelPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:信道为无效信道" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }
            
            channelPtr->queueCancelConsume(req);
            LOG(logLevel::DEBUG) << "执行完毕" << std::endl;
        }

        //发布消息
        void onBasicPublish(const muduo::net::TcpConnectionPtr& conn,const basicPublishRequestPtr& req,muduo::Timestamp)
        {
            //从链接管理模块中获取特定链接
            Connection::ptr connPtr = m_connections->getConnection(conn);
            
            //判断链接是否为有效链接
            if(connPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:连接为无效连接" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }

            Channel::ptr channelPtr = connPtr->getChannel(req->channel_id());

            //判断信道是否为有效信道
            if(channelPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:信道为无效信道" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }
            
            channelPtr->basicPublish(req);
            LOG(logLevel::DEBUG) << "执行完毕" << std::endl;
        }

        //确认消息
        void onBasicAck(const muduo::net::TcpConnectionPtr& conn,const basicAckRequestPtr& req,muduo::Timestamp)
        {
            //从链接管理模块中获取特定链接
            Connection::ptr connPtr = m_connections->getConnection(conn);
            
            //判断链接是否为有效链接
            if(connPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:连接为无效连接" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }

            Channel::ptr channelPtr = connPtr->getChannel(req->channel_id());

            //判断信道是否为有效信道
            if(channelPtr.get() == nullptr)
            {
                LOG(logLevel::ERROR) << "执行失败:信道为无效信道" << std::endl;
                //关闭传入的链接
                conn->shutdown();
                return;
            }
            
            channelPtr->basicAck(req);

            LOG(logLevel::DEBUG) << "执行完毕" << std::endl;
        }

        /////////////////////////////////////////////////////////////////////////////////
        ///启动服务器
        /////////////////////////////////////////////////////////////////////////////////
        void start()
        {
            m_server.start();
            m_loop.loop();
        }

        //服务器未知消息处理逻辑
        void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn, const MessagePtr& message, muduo::Timestamp) 
        {
            LOG(logLevel::ERROR) << "服务端收到未知类型的消息" << std::endl;
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }

        //服务器连接到来的处理逻辑
        void onConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            if(conn->connected())
            {
                m_connections->createConnection(conn,m_codec,m_pool,m_host,m_consumers);
            }
            else
            {
                m_connections->removeConnection(conn);
            }
        }

    private:
        muduo::net::EventLoop m_loop; //事件循环
        muduo::net::TcpServer m_server; //服务器
        ProtobufDispatcher m_dispatcher; //事件分发器
        ProtobufCodecPtr m_codec; //协议处理器
        VirtualHost::ptr m_host; //虚拟机管理
        ConsumerManager::ptr m_consumers; //消费者管理
        ConnectionManager::ptr m_connections; //连接管理
        threadpool::ptr m_pool;
    };

};



#endif 